package com.bs.kafka.listener;

import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson.JSON;
import lombok.extern.log4j.Log4j2;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.ProducerListener;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;

/**
 * @author bingshao
 * @date 2022/10/13
 **/
@Configuration
@Log4j2
public class KafkaListener {

    @Resource
    private KafkaTemplate<String, Object> kafkaTemplate;

    @Resource
    private RedisTemplate<String, Object> redisTemplate;

    /**
     * 配置监听
     */
    @PostConstruct
    private void listener() {
        kafkaTemplate.setProducerListener(new ProducerListener<String, Object>() {

            @Override
            public void onSuccess(ProducerRecord<String, Object> producerRecord, RecordMetadata recordMetadata) {
                String topic = recordMetadata.topic();
                int partition = recordMetadata.partition();
                long offset = recordMetadata.offset();
                log.info("成功====》主题：" + topic + "=====>分区:" + partition + "=====>位移:" + offset);

                // 成功则删除持久化的消息
                String key = producerRecord.key();
                log.info("key======={},", key);
                if (StrUtil.isNotBlank(key)) {
                    redisTemplate.delete(key);
                }
            }

            @Override
            public void onError(ProducerRecord<String, Object> producerRecord, RecordMetadata recordMetadata, Exception exception) {
                log.error("错误原因=============={}", exception.getMessage());

                // 失败则可以重新发送，一般不会失败
                String key = producerRecord.key();
                log.info("key======={},", key);
                if (StrUtil.isNotBlank(key)) {
                    Object value = redisTemplate.opsForValue().get(key);
                    if (value != null) {
                        String message = JSON.toJSONString(value);
                        kafkaTemplate.send(recordMetadata.topic(), key, message);
                    }
                }
            }
        });

    }


}
